package org.fastsyncer.worker.framework;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;

import org.fastsyncer.common.entity.MappingTask;
import org.fastsyncer.parser.framework.ParserFactory;
import org.json.JSONArray;
import org.json.JSONException;
import org.json.JSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Java 线程池 Java通过Executors提供四种线程池，分别为：
 * newCachedThreadPool创建一个可缓存线程池，如果线程池长度超过处理需要，可灵活回收空闲线程，若无可回收，则新建线程。
 * newFixedThreadPool 创建一个定长线程池，可控制线程最大并发数，超出的线程会在队列中等待。 newScheduledThreadPool
 * 创建一个定长线程池，支持定时及周期性任务执行。 newSingleThreadExecutor
 * 创建一个单线程化的线程池，它只会用唯一的工作线程来执行任务，保证所有任务按照指定顺序(FIFO, LIFO, 优先级)执行。
 * 
 * @author YangPei
 * @date 2017年5月13日 下午7:16:37
 * @version 1.0.0
 */
public class WorkerThreadPool {

	private static Logger logger = LoggerFactory.getLogger(WorkerThreadPool.class);

	/**
	 * CountDownLatch，一个同步辅助类，在完成一组正在其他线程中执行的操作之前，它允许一个或多个线程一直等待。
	 */
	private CountDownLatch latch;
	
	/**
	 * 存储所有Task,每个Task里面包含执行的任务
	 */
	private ConcurrentLinkedQueue<Integer[]> workerQuene;

	/**
	 * 记录执行成功和失败数
	 */
	private ConcurrentHashMap<String, AtomicInteger> sync = new ConcurrentHashMap<>();

	/**
	 * 失败原因
	 */
	private ConcurrentLinkedQueue<JSONObject> reasonQuene = new ConcurrentLinkedQueue<>();

	/**
	 * 处理同步任务
	 * 
	 * @param mappingTask
	 *            同步任务
	 * @param tasks
	 *            任务清单,[{0,999},{1000,1999},...]
	 * @throws InterruptedException
	 * @throws JSONException
	 */
	public JSONObject sync(MappingTask mappingTask, ConcurrentLinkedQueue<Integer[]> tasks)
			throws InterruptedException, JSONException {
		// 获取要执行的任务
		workerQuene = tasks;

		// 线程个数
		latch = new CountDownLatch(mappingTask.getThreadNum());

		sync.clear();
		reasonQuene.clear();
		// 设置执行总数和成功数
		sync.put("excute", new AtomicInteger(0));
		sync.put("success", new AtomicInteger(0));

		// 创建线程池
		ExecutorService es = Executors.newCachedThreadPool();
		// 分配配置中设置的线程数
		for (int i = 0; i < mappingTask.getThreadNum(); i++) {
			es.execute(new Runnable() {

				@Override
				public void run() {
					while (!workerQuene.isEmpty()) {
						// 消费
						Integer[] pos = workerQuene.poll();
						try {
							Integer[] result = ParserFactory.getInstance().parse(mappingTask, pos[0], pos[1]);
							// 累加执行总数、成功条数
							sync.get("excute").getAndAdd(result[0]);
							sync.get("success").getAndAdd(result[1]);
						} catch (Exception e) {
							try {
								// 记录错误信息
								JSONObject err = new JSONObject();
								if(pos != null && pos.length == 2){
									err.put("startPos", pos[0]);
									err.put("endPos", pos[1]);
								}
								err.put("cause", e.getCause());
								err.put("message", e.getMessage());
								reasonQuene.offer(err);
							} catch (JSONException e1) {
								logger.error("记录错误信息时：转换JSONObject异常！");
							}
						}
					}
					latch.countDown();
				}

			});
		}
		// 使得主线程(main)阻塞直到latch.countDown()为零才继续执行
		latch.await();
		es.shutdown();

		// 统计错误原因，因为考虑产生的错误不会很多，size不会影响性能
		JSONArray reasonArr = new JSONArray();
		int i = 0;
		while (!reasonQuene.isEmpty()) {
			JSONObject err = reasonQuene.poll();
			reasonArr.put(i, err);
			logger.info("记录错误(" + (i + 1) + ")：" + err);
			i++;
		}

		// 统计执行结果
		JSONObject result = new JSONObject();
		result.put("excute", sync.get("excute").get());
		result.put("success", sync.get("success").get());
		result.put("reason", reasonArr);
		return result;
	}

}
